package mr

import (
	"fmt"
	"hash/fnv"
	"io/ioutil"
	"log"
	"net/rpc"
	"os"
	"time"

	"encoding/json"
	"sort"
	"strconv"
)

//
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}

// for sorting by key.
type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

//
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	// Your worker implementation here.
	for {
		dispathcArgs := DispatchTaskArgs{}
		dispatchReply := DispatchTaskReply{}
		ok := call("Coordinator.DispatchTask", &dispathcArgs, &dispatchReply)
		if !ok {
			log.Fatalln("call Coordinator.DispatchTask failed")
		}

		log.Println("worker fetch a task, id =", dispatchReply.TaskId)

		if dispatchReply.TaskId == -1 { // 暂时无任务，等待1s
			time.Sleep(time.Second)
			continue
		}
		if dispatchReply.TaskType == kMapTaskType {
			mapTask(&dispatchReply, mapf)
		} else {
			reduceTask(&dispatchReply, reducef)
		}
	}
}

func mapTask(dispatchReply *DispatchTaskReply, mapf func(string, string) []KeyValue) {
	log.Println("handle a map task")

	nReduce := dispatchReply.NReduce

	// 获取inputfile name
	log.Println("get inputfile name") // for debug
	filename, ok := dispatchReply.File.(string)
	if !ok {
		log.Fatalln("in func mapTask, dispatchReply.File.(string) error")
	}

	// 读入inputfile
	log.Println("read inputfile") // for debug
	file, err := os.Open(filename)
	if err != nil {
		log.Fatalf("cannot open %v", filename)
	}
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Fatalf("cannot read %v", filename)
	}
	file.Close()

	// call map
	log.Println("call map") // for debug
	kva := mapf(filename, string(content))

	// 创建intermediate file
	log.Println("create intermediate file") // for debug
	ofiles := make([]*os.File, nReduce)
	encs := make([]*json.Encoder, nReduce)
	for i := 0; i < nReduce; i++ {
		oname := "intermediate/mr-" + strconv.Itoa(dispatchReply.TaskId) + "-" + strconv.Itoa(i)
		ofiles[i], _ = os.OpenFile(oname, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0666)
		encs[i] = json.NewEncoder(ofiles[i])
	}

	vis := make([]bool, nReduce)
	// 将map输出写入各个文件
	log.Println("output intermdeiate") // for debug
	for i := range kva {
		reduceId := ihash(kva[i].Key) % nReduce
		err = encs[reduceId].Encode(kva[i])
		vis[reduceId] = true
	}

	// 关闭文件
	log.Println("close files") // for debug
	for i := range ofiles {
		ofiles[i].Close()
	}

	// 向coordinaotr报告任务完成
	log.Println("before call Coordinator.FinishTask") // for debug
	finishArgs := FinishTaskArgs{}
	finishReply := FinishTaskReply{}

	intermediates := make([]string, nReduce)
	for i, exist := range vis {
		if exist {
			intermediates[i] = "intermediate/mr-" + strconv.Itoa(dispatchReply.TaskId) + "-" + strconv.Itoa(i)
		}
	}
	finishArgs.TaskType = kMapTaskType
	finishArgs.TaskId = dispatchReply.TaskId
	finishArgs.IntermediateFiles = intermediates

	ok = call("Coordinator.FinishTask", &finishArgs, &finishReply)
	if !ok {
		log.Fatal("in map task, call Coordinator.FinishTask fail")
	}

	log.Println("finish a map task")
}

func reduceTask(dispatchReply *DispatchTaskReply, reducef func(string, []string) string) {
	log.Println("handle a reduce task")

	filenames, ok := dispatchReply.File.([]string)
	if !ok {
		log.Fatalln("in func reduceTask, dispatchReply.File.([]string) error")
	}
	reduceId := dispatchReply.TaskId

	// 读入这一组中间文件的所有kv
	intermediate := []KeyValue{}
	for _, filename := range filenames {
		file, err := os.OpenFile(filename, os.O_RDONLY, 0666)
		dec := json.NewDecoder(file)
		for {
			var kv KeyValue
			if err = dec.Decode(&kv); err != nil {
				break
			}
			intermediate = append(intermediate, kv)
		}
	}

	sort.Sort(ByKey(intermediate))

	oname := "mr-out-" + strconv.Itoa(reduceId)
	ofile, _ := os.Create(oname)

	//
	// call Reduce on each distinct key in intermediate[],
	// and print the result to mr-out-0.
	//
	i := 0
	for i < len(intermediate) {
		j := i + 1
		for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
			j++
		}
		values := []string{}
		for k := i; k < j; k++ {
			values = append(values, intermediate[k].Value)
		}
		output := reducef(intermediate[i].Key, values)

		// this is the correct format for each line of Reduce output.
		fmt.Fprintf(ofile, "%v %v\n", intermediate[i].Key, output)

		i = j
	}

	ofile.Close()

	// 报告任务完成
	finishArgs := FinishTaskArgs{}
	finishReply := FinishTaskReply{}

	finishArgs.TaskType = kReduceTaskType
	finishArgs.TaskId = reduceId

	ok = call("Coordinator.FinishTask", &finishArgs, &finishReply)
	if !ok {
		log.Fatal("in reduce task, call Coordinator.FinishTask fail")
	}

	log.Println("finish a reduce task")
}

//
// example function to show how to make an RPC call to the coordinator.
//
// the RPC argument and reply types are defined in rpc.go.
//
func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 99

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	// the "Coordinator.Example" tells the
	// receiving server that we'd like to call
	// the Example() method of struct Coordinator.
	ok := call("Coordinator.Example", &args, &reply)
	if ok {
		// reply.Y should be 100.
		fmt.Printf("reply.Y %v\n", reply.Y)
	} else {
		fmt.Printf("call failed!\n") ///////////////////////////////////////////这里可能要增加exit
	}
}

//
// send an RPC request to the coordinator, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := coordinatorSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
